package com.gemantic.task.runner;


import cn.hutool.core.collection.CollectionUtil;
import lombok.NonNull;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.Assert;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gemantic.task.dao.constant.TaskStatus;
import com.gemantic.task.dao.repository.BaseTaskRepository;
import com.gemantic.task.dto.ResResult;
import com.gemantic.task.dto.ResultAction;
import com.gemantic.task.dto.TaskContext;
import com.gemantic.task.handler.BaseTaskHandler;
import com.gemantic.task.handler.ITaskHandler;
import com.gemantic.task.model.BaseTask;
import com.gemantic.task.model.Task;
import com.gemantic.task.util.DateUtil;
import com.gemantic.task.util.SpringUtils;
import com.gemantic.task.util.TaskUtil;
import com.gemantic.task.util.ThreadPoolUtil;

/**
 * @author xiedi
 * @date 2022/09/15
 **/
public abstract class BaseTaskRunner<T extends BaseTask> {
    protected static final Logger LOGGER = LoggerFactory.getLogger(BaseTaskRunner.class);

    private static final int MEMO_MAX_LENGTH = 2000;
    private static final int MAX_INTERVAL_SECOND = 3600 * 24;
    private static final int POSTPONE_MINUTES = 30;

    /**
     * 执行任务
     *
     * @param taskArg 待执行任务，只需要传入id即可
     * @param args    执行需要的扩展参数，目前需要runnerName
     * @return 任务链最后task的结果。
     */
    public  ResResult run(T taskArg, Map<String, String> args) {

        if (args == null) {
            args = new HashMap<>();
        }

        ResResult resResult = assertParam(taskArg, args);
        if (resResult != null) {
            logException(null, null, new Exception(resResult.getMsg()));
            return resResult;
        }

        T task = getTaskRepository().getTask(taskArg, args);
        if (null == task) {
            return ResResult.process(String.format("不能找到Task{%d}.", taskArg.getPrimaryKey()), null);
        }
        return this.innerRun(task, args);
    }

    protected ResResult assertParam(T taskArg, Map<String, String> args) {
        if (taskArg == null) {
            return ResResult.process("taskhandler is null.", null);
        }
        if (taskArg.getPrimaryKey() < 1) {
            return ResResult.process(String.format("taskhandler id  %s ", taskArg.getPrimaryKey()), null);
        }
        return null;
    }

    private ResResult innerRun(T task, Map<String, String> args) {
        Object taskLog = null;
        ResResult result = null;
        ITaskHandler taskHandler = null;
        TaskContext<T> taskContext = null;
        try {
            taskContext = new TaskContext<>(task);
            //验证task 状态
            result = this.validTask(task, false);

            if (null != result) {
                return result;
            }

            //3.获取TaskHandler
            taskHandler = getTaskHandler(task);
            if (null == taskHandler) {
                return ResResult.fail("系统错误，不能找到taskhandler.");
            }
            taskLog = startTaskLog(task, args);

            // 修改taskStatus 为running
            result = this.runningTask(taskContext, taskHandler);
            if (result != null) {
                return result;
            }
            result = this.executeTask(taskContext, taskHandler);

        } catch (Exception ex) {
            LOGGER.error("执行TASK出错", ex);
            result = ResResult.exception(ExceptionUtils.getMessage(ex), ex);

        } finally {
            endTaskLog(taskLog, task, result);
            notice(task, taskHandler, result);
        }

        ResResult afterResult = afterProcess(taskContext, taskHandler, result);
        return afterResult != null ? afterResult : result;
    }

    @SuppressWarnings("unchecked")
    private void notice(T task, ITaskHandler taskHandler, ResResult result) {
        if (result != null
                && result.getActions().contains(ResultAction.NOTICE)) {

            String noticeMsg = String.format("%s%s%s",
                    result.getMsg(),
                    System.lineSeparator(),
                    result.getException() != null ? result.getException() : StringUtils.EMPTY);

            String content = buildNoticeMsg(task, taskHandler, noticeMsg);
            noticeMsg(noticeMsg.hashCode(), content);
        }
    }

    private void noticeMsg(int msgID, String message) {
        //TODO 异常时告警
    }

    private String getLoggerCategory(T task, ITaskHandler taskHandler) {
        String category = this.getClass().toString();

        if (taskHandler != null) {
            category = taskHandler.getClass().toString();
        } else if (task != null) {
            category = task.getType();
        }
        return category;
    }

    private ResResult executeTask(TaskContext<T> taskContext, ITaskHandler taskHandler) {
        Object processData = this.getProcessData(taskContext.getCurrentTask(), taskHandler);
        taskContext.setReqObjectData(processData);
        ResResult result = beforeProcess(taskContext, processData, taskHandler);
        if (null == result) {
            result = tranProcess(taskContext, processData, taskHandler);
        }
        return result;
    }

    protected ResResult handleTaskResult(TaskContext<T> taskContext, ITaskHandler taskHandler, ResResult result, Exception ex) {

        T task = taskContext.getCurrentTask();

        buildTaskResponse(task, result, ex);

        saveTask(taskContext, taskHandler, result);

        return result;
    }

    protected void saveTask(TaskContext<T> taskContext, ITaskHandler taskHandler, ResResult result) {
        if (result.getActions().contains(ResultAction.NEED_POSTPONE)) {
            postponeTask(taskContext, result);
        } else if (result.getActions().contains(ResultAction.NEED_RETRY)) {
            retryTask(taskContext, taskHandler, result);
        } else {
            getTaskRepository().closeTask(taskContext.getCurrentTask());
        }
    }

    private void retryTask(TaskContext<T> taskContext, ITaskHandler taskHandler, ResResult result) {
        Date nextRunDate = taskHandler.getNextRunTime(taskContext, result);
        Date cutOffTime = null;
        int maxRunTime = taskHandler.getMaxRunTime(taskContext.getReqObjectData());
        if (maxRunTime > 0) {
            cutOffTime = DateUtil.addSeconds(taskContext.getCreateTime(), maxRunTime);
        }
        boolean cutOff = cutOffTime != null && cutOffTime.before(DateUtil.now());
        if (taskContext.getRetryTimes() < taskContext.getMaxRetryTimes() && nextRunDate != null  && !cutOff) {
            getTaskRepository().retryTask(taskContext.getCurrentTask(), nextRunDate);
        } else {
            getTaskRepository().terminateTask(taskContext.getCurrentTask());
        }
    }

    private void postponeTask(TaskContext<T> taskContext, ResResult result) {
        Date nextRunDate = DateUtil.addMinutes(POSTPONE_MINUTES);
        if ( result.getData() instanceof Date ) {
            nextRunDate = (Date) result.getData();
            nextRunDate = nextRunDate.before(DateUtil.now()) ? DateUtil.addMinutes(1) : nextRunDate;
            nextRunDate = nextRunDate.after(DateUtil.addSeconds(DateUtil.now(), TaskUtil.getMaxIntervalSeconds()))
                    ? DateUtil.addSeconds(DateUtil.now(), MAX_INTERVAL_SECOND) : nextRunDate;
        }
        getTaskRepository().postponeTask(taskContext.getCurrentTask(), nextRunDate);
    }

    protected void buildTaskResponse(T task, ResResult result, Exception ex) {
        task.setResponseData(JSON.toJSONString(result));
        String memo;
        if (ex != null) {
            memo = ExceptionUtils.getMessage(ex);
        } else {
            memo = StringUtils.trimToEmpty(result.getMsg());
        }
        if (memo.length() > MEMO_MAX_LENGTH) {
            memo = memo.substring(0, MEMO_MAX_LENGTH);
        }
        task.setRemark(memo);
    }

    @SuppressWarnings("unchecked")
    private ResResult tranProcess(TaskContext<T> taskContext, Object processData, ITaskHandler taskHandler) {
        ResResult result;
        PlatformTransactionManager transactionManager = null;
        TransactionStatus tran = null;
        T task = taskContext.getCurrentTask();

        try {

            transactionManager = (PlatformTransactionManager) SpringUtils.getBean("transactionManager");
            DefaultTransactionDefinition transactionDefinition =
                    new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRED);
            transactionDefinition.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
            tran = transactionManager.getTransaction(transactionDefinition);
            T dbTask = lockTask(task);
            if (null == dbTask) {
                rollbackTransaction(transactionManager, tran);
                return ResResult.fail("tranProcess执行任务失败");
            }
            result = validTask(dbTask, true);
            if (null != result) {
                rollbackTransaction(transactionManager, tran);
                return result;
            }
            task = dbTask;
            result = taskHandler.process(processData, taskContext);
            if (result == null) {
                throw new RuntimeException("不能获取到处理结果");
            }
            result = handleTaskResult(taskContext, taskHandler, result, null);

            transactionManager.commit(tran);
        } catch (Exception ex) {
            rollbackTransaction(transactionManager, tran);
            result = handleProcessException(taskContext, taskHandler, ex);

        } finally {
            rollbackTransaction(transactionManager, tran);
        }
        Assert.notNull(result, "taskhandler handler process 未返回值，必须返回ResResult.");
        return result;
    }

    protected Object getProcessData(T task, ITaskHandler handler) {
        ParameterizedType parameterizedType = (ParameterizedType) handler.getClass().getGenericSuperclass();
        Type type = parameterizedType.getActualTypeArguments()[0];
        return JSONObject.parseObject(task.getRequestData(), type);
    }

    private ResResult handleProcessException(TaskContext<T> taskContext, ITaskHandler taskHandler, Exception ex) {
        ResResult result = taskHandler.handleException(taskContext, ex);
        if (null == result) {
            logException(taskContext.getCurrentTask(), taskHandler, ex, false);
            result = ResResult.exception(String.format("系统错误:%s", ExceptionUtils.getStackTrace(ex)), ex);
        }
        try {
            handleTaskResult(taskContext, taskHandler, result, ex);
        } catch (Exception innerEx) {
            logException(taskContext.getCurrentTask(), taskHandler, innerEx, false);
            result = ResResult.exception(String.format("系统错误:%s", ExceptionUtils.getStackTrace(innerEx)), innerEx);
        }
        result.setException(ex);
        return result;
    }

    protected void logException(T task, ITaskHandler taskHandler, Exception ex) {
        logException(task, taskHandler, ex, true);
    }

    protected void logException(T task, ITaskHandler taskHandler, Exception ex, boolean needNotice) {
        String msg = buildNoticeMsg(task, taskHandler, ExceptionUtils.getMessage(ex));
        LOGGER.error(getLoggerCategory(task, taskHandler), msg, ex);
        if (needNotice) {
            //TODO 发送微信报警信息
        }
    }

    private String buildNoticeMsg(T task, ITaskHandler taskHandler, String message) {
        StringBuilder msgbuilder = new StringBuilder();
        if (taskHandler != null) {
            msgbuilder.append(taskHandler.getHandlerName(new TaskContext<>(task)));
            msgbuilder.append(",");
        }
        if (task != null) {
            msgbuilder.append("Id:");
            msgbuilder.append(task.getPrimaryKey());
            msgbuilder.append(",orderno:");
            msgbuilder.append(task.getOrderNo());
            msgbuilder.append(System.lineSeparator());
        }
        msgbuilder.append(message);
        return msgbuilder.toString();
    }

    private T lockTask(T task) {
        return getTaskRepository().lockTask(task);
    }

    private ResResult runningTask(@NonNull TaskContext<T> taskContext, @NonNull ITaskHandler taskHandler) {
        try {
            int delayMinutes = taskHandler.estimateRunMinutes();
            delayMinutes = Math.max(BaseTaskHandler.ESTIMATE_RUN_MINUTE_SHORT, delayMinutes);
            T tmpTask = getTaskRepository().runningTask(taskContext.getCurrentTask(), delayMinutes);
            if (tmpTask == null) {
                return ResResult.process("任务执行中，请稍等", null);
            }
        } catch (Exception ex) {
            LOGGER.error("执行任务报错", ex);
            return ResResult.exception("系统异常，请重试", ex);
        }
        return null;
    }

    private ResResult beforeProcess(TaskContext<T> taskContext, Object processData, ITaskHandler taskHandler) {

        ResResult result = needPostpone(taskContext, processData, taskHandler);
        if (result == null) {
            try {
                result = taskHandler.beforeProcess(processData, taskContext);
            } catch (Exception ex) {
                return handleProcessException(taskContext, taskHandler, ex);
            }
        }

        if (result != null) {
            try {
                return handleTaskResult(taskContext, taskHandler, result, null);
            } catch (Exception ex) {
                logException(taskContext.getCurrentTask(), taskHandler, ex, false);
                return ResResult.exception(String.format("系统内部异常：%s", ExceptionUtils.getMessage(ex)), ex);
            }
        }
        return null;
    }

    private ResResult needPostpone(TaskContext<T> taskContext, Object processData, ITaskHandler taskHandler) {

        try {
            return taskHandler.needPostpone(processData, taskContext);
        } catch (Exception ex) {
            LOGGER.error(String.format("调用挂起异常，延迟执行.%s", ExceptionUtils.getMessage(ex)), ex);
            return ResResult.suspend("调用挂起异常，延迟执行");
        }
    }

    private ResResult afterProcess(TaskContext<T> taskContext, ITaskHandler taskHandler, ResResult result) {

        ResResult afterResult = null;

        // 1.优先执行taskhandler的afterProcess
        if (taskHandler != null) {
            try {
                afterResult = taskHandler.afterProcess(taskContext);
            } catch (Exception ex) {
                LOGGER.error("处理任务异常", ex);
                afterResult = ResResult.process(ex.getMessage(), null);
            }
        }
        // 2.处理超出最大重试次数的逻辑
        ResResult maxRetryTimesResult = this.afterMaxRetryTimes(taskContext, taskHandler);
        if (maxRetryTimesResult != null) {
            afterResult = maxRetryTimesResult;
        }
        // 3.处理超出最大重试时间的逻辑
        ResResult afterMaxRunTimeResult = this.afterMaxRunTime(taskContext, taskHandler);
        if (afterMaxRunTimeResult == null) {
            afterResult = afterMaxRunTimeResult;
        }
        // 4.在执行异步后台任务.
        executeBackendTask(taskContext);

        // 5.在执行下一任务
        ResResult nextResResult = executeNextTask(taskContext);

        return nextResResult != null ? nextResResult : afterResult;
    }

    private ResResult afterMaxRetryTimes(TaskContext<T> taskContext, ITaskHandler taskHandler) {
        int retryTimes = taskContext.getRetryTimes() == null ? 0 : taskContext.getRetryTimes();
        int maxRetryTimes = taskContext.getMaxRetryTimes();
        if (maxRetryTimes <= retryTimes) {
            return taskHandler.afterMaxRetryTimes(taskContext.getReqObjectData(), taskContext);
        }
        return null;
    }

    private ResResult afterMaxRunTime(TaskContext<T> taskContext, ITaskHandler taskHandler) {
        int maxRunTime = taskHandler.getMaxRunTime(taskContext.getReqObjectData());
        if (maxRunTime > 0 && taskContext.getCreateTime() != null) {
            Date terminateTime = DateUtil.addSeconds(taskContext.getCreateTime(), maxRunTime);
            if (terminateTime.before(DateUtil.now())) {
                return taskHandler.afterMaxRunTime(taskContext.getReqObjectData(), taskContext);
            }
        }
        return null;
    }


    private ResResult executeNextTask(TaskContext<T> taskContext) {
        if (taskContext == null) {
            return null;
        }
        ResResult resResult = null;

        String fromRunner = String.format("from %s%s ",
                taskContext.getCurrentTask().getClass().getSimpleName(),
                taskContext.getCurrentTask().getPrimaryKey());

        BaseTask nextTask = taskContext.pollNextTask();
        while (nextTask != null) {

            resResult = executeBaseTask(nextTask, fromRunner);

            nextTask = taskContext.pollNextTask();
        }


        return resResult;
    }

    private ResResult executeBaseTask(BaseTask baseTask, String runnerName) {
        try {
            if (baseTask == null) {
                return null;
            }
            Map<String, String> args = TaskUtil.createTaskArgs(runnerName);
            if (baseTask instanceof Task) {
                TaskRunner runner = SpringUtils.getBean(TaskRunner.class);
                return runner.run((Task) baseTask, args);

            }
            throw new IllegalArgumentException("只能Task能异步后台执行");
        } catch (Exception ex) {
            LOGGER.error(ExceptionUtils.getStackTrace(ex));
        }
        return null;
    }


    private void executeBackendTask(TaskContext taskContext) {
        if (taskContext == null) {
            return;
        }

        String fromRunner = String.format("from %s%s ",
                taskContext.getCurrentTask().getClass().getSimpleName(),
                taskContext.getCurrentTask().getPrimaryKey());

        try {
            final List<BaseTask> backendTasks = taskContext.pollAllBackendTask();
            if (!CollectionUtil.isEmpty(backendTasks)) {
                for (BaseTask baseTask : backendTasks) {
                    ThreadPoolUtil.execute(() -> executeBaseTask(baseTask, fromRunner));
                }
            }
        } catch (Exception ex) {
            LOGGER.error(ExceptionUtils.getMessage(ex));
        }
    }


    private Object startTaskLog(T task, Map<String, String> args) {
        try {
            return args;
        } catch (Exception ex) {
            LOGGER.error(ExceptionUtils.getStackTrace(ex));
            return null;
        }
    }


    private void endTaskLog(Object taskLog, T task, ResResult resResult) {
        if (taskLog != null) {
            LOGGER.info("taskLog : [{}], task id: [{}], task info: [{}]", taskLog, task.getOrderNo(), resResult.getMsg());
        }
    }

    private void rollbackTransaction(PlatformTransactionManager transactionManager, TransactionStatus tran) {
        if (tran != null
                && transactionManager != null
                && !tran.isCompleted()) {
            try {
                transactionManager.rollback(tran);
            } catch (Exception ex) {
                logException(null, null, ex);
            }
        }
    }

    protected abstract BaseTaskRepository<T> getTaskRepository();


    protected abstract ITaskHandler getTaskHandler(T task);

    protected ResResult validTask(T task, boolean ignoreRunning) {
        ResResult resResult = null;
        if (TaskStatus.CLOSE.isEquals(task.getStatus())) {
            resResult = parseResponse(task, ResResult.success());
        } else if (TaskStatus.TERMINATED.isEquals(task.getStatus())) {
            resResult = parseResponse(task, ResResult.process("任务终止", null));
        } else if (!ignoreRunning
                && TaskStatus.RUNNING.isEquals(task.getStatus())
                && task.getNextRunTime().after(DateUtil.now())) {
            resResult = ResResult.process("任务正在执行，请稍后重试", null);
        }
        return resResult;
    }

    protected ResResult parseResponse(T task, ResResult defaultValue) {
        String response = task.getResponseData();
        if (!StringUtils.isEmpty(response)) {
            return JSONObject.parseObject(response, ResResult.class);
        }
        return defaultValue;
    }
}
